package com.my.control;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.my.entity.ResultMessage;
import com.my.service.task.log.BuyCartProductLog;
import com.my.service.task.log.CollectProductLog;
import com.my.service.task.log.FollowProductLog;
import com.my.service.task.log.ScanProductLog;
import com.my.service.task.utils.ReadProperties;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.Date;

@RestController
@RequestMapping(value = "infolog", method = RequestMethod.GET)
class InfoInControl{

    // 读取kafka配置
    private static final String propertyFileName = "InfoInService.properties";
    private final String buyCartProductLogTopic = ReadProperties.getKey("buyCartProductLog", propertyFileName);
    private final String collectProductLogTopic = ReadProperties.getKey("collectProductLog", propertyFileName);
    private final String followProductLogTopic = ReadProperties.getKey("followProductLog", propertyFileName);
    private final String scanProductLogTopic = ReadProperties.getKey("scanProductLog", propertyFileName);

    // kafka传输工具类
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @RequestMapping(value = "helloworld", method = RequestMethod.GET)
    public String helloworld(HttpServletRequest req){
        String ip = req.getRemoteAddr();
        ResultMessage resultMessage = new ResultMessage();
        resultMessage.setMessage("Hello Wencongliu!" + ip);
        resultMessage.setStatus("success");
        return JSON.toJSONString(resultMessage);
    }

    /**
     * BuyCartProductLog: {productId:productId}...
     * CollectProductLog: {productId:productId}...
     * FollowProductLog: {productId:productId}...
     * ScanProductLog: {productId:productId}...
     * @param receivelog
     * @return
     */

    @RequestMapping(value = "receivelog", method = RequestMethod.POST)
    public String helloworld(String receivelog){
        if(StringUtils.isBlank(receivelog)){
            return null;
        }
        String[] receiveArrays = receivelog.split(":", 2);
        String classname = receiveArrays[0];
        String data = receiveArrays[1];
        String resultMessage = "";
        if(classname.equals("BuyCartProductLog")){
            BuyCartProductLog buyCartProductLog = JSONObject.parseObject(data, BuyCartProductLog.class);
            resultMessage = JSONObject.toJSONString(buyCartProductLog);
            kafkaTemplate.send(buyCartProductLogTopic,resultMessage + "##1##" + new Date().getTime());
        }
        else if(classname.equals("CollectProductLog")){
            CollectProductLog collectProductLog = JSONObject.parseObject(data, CollectProductLog.class);
            resultMessage = JSONObject.toJSONString(collectProductLog);
            kafkaTemplate.send(collectProductLogTopic,resultMessage + "##1##" + new Date().getTime());
        }
        // 对于关注的产品信息 返回的字符串是由 产品本身的信息 + 单位1 + 日期的时间戳数字
        else if(classname.equals("FollowProductLog")){
            FollowProductLog followProductLog = JSONObject.parseObject(data, FollowProductLog.class);
            resultMessage = JSONObject.toJSONString(followProductLog);
            kafkaTemplate.send(followProductLogTopic,resultMessage + "##1##" + new Date().getTime());
        }
        else if(classname.equals("ScanProductLog")){
            ScanProductLog scanProductLog = JSONObject.parseObject(data, ScanProductLog.class);
            resultMessage = JSONObject.toJSONString(scanProductLog);
            kafkaTemplate.send(scanProductLogTopic,resultMessage + "##1##" + new Date().getTime());
        }
        ResultMessage res = new ResultMessage();
        res.setMessage(resultMessage);
        res.setStatus("success");
        return JSON.toJSONString(res);
    }


}